//==============================================================================
// Copyright (c) 2008-2013 Niall Ryan. All Rights Reserved.
//==============================================================================

#ifndef HYDRA_EXCHANGERS_LOCKFREEEXCHANGER_H
#define HYDRA_EXCHANGERS_LOCKFREEEXCHANGER_H 1

#include "hydra/atomic/AtomicPair.h"

namespace Hydra
{
	/**
	 * An exchanger allows threads to rendezvous and exchange items.
	 */
	template<typename T, unsigned int MAX_ITERATIONS=100>
	class LockFreeExchanger
	{
	public:
		LockFreeExchanger();

		/**
		 * Attempts to exchange an item with another thread, this can only rendezvous with another exchange function.
		 */
		bool exchange(const T& item, T* resultItem);

		/**
		 * Offers an item to another thread, this can only rendezvous with an accept function.
		 */
		bool offer(const T& item);

		/**
		 * Attempts to accept an item from another thread, this can only rendezvous with an offer function.
		 */
		bool accept(T* resultItem);

	private:
		enum State
		{
			STATE_EMPTY,
			STATE_EXCHANGE_WAITING,
			STATE_OFFER_WAITING,
			STATE_BUSY
		};
		AtomicPair<T, State> m_slot;
	};

	//==================================================================================================
	//==================================================================================================
	//==================================================================================================

	template<typename T, unsigned int MAX_ITERATIONS>
	LockFreeExchanger<T, MAX_ITERATIONS>::LockFreeExchanger()
	{
		m_slot.set(T(), STATE_EMPTY);
	}

	template<typename T, unsigned int MAX_ITERATIONS>
	bool LockFreeExchanger<T, MAX_ITERATIONS>::exchange(const T& item, T* resultItem)
	{
		unsigned int iterationCount = 0;
		while( iterationCount < MAX_ITERATIONS )
		{
			T slotItem;
			State slotState;
			m_slot.get(&slotItem, &slotState);
			switch( slotState )
			{
			case STATE_EMPTY:
				{
					//slot is empty, so attempt to offer our item for an exchange
					if( m_slot.compareAndSwap(item, STATE_EXCHANGE_WAITING, slotItem, STATE_EMPTY) )
					{
						//wait for another thread to rendezvous and exchange items
						unsigned int waitIterationCount = 0;
						while( waitIterationCount < MAX_ITERATIONS )
						{
							m_slot.get(&slotItem, &slotState);
							if( slotState == STATE_BUSY )
							{
								//exchange was successful, return the item from the other thread
								//no compareAndSwap needed here, we're the only thread that can change the state now
								m_slot.set(T(), STATE_EMPTY);
								*resultItem = slotItem;
								return true;
							}
							++waitIterationCount;
						}
						//time's up, attempt to cancel our exchange
						if( m_slot.compareAndSwap(T(), STATE_EMPTY, item, STATE_EXCHANGE_WAITING) )
						{
							//cancelled successfully, return since we've already spent all the iterations available
							return false;
						}
						else
						{
							//cancel failed, that means the exchange must have succeeded at the last moment
							m_slot.get(&slotItem, &slotState);
							assert(slotState == STATE_BUSY);
							*resultItem = slotItem;
							m_slot.set(T(), STATE_EMPTY);
							return true;
						}
					}
				}
				break;
			case STATE_EXCHANGE_WAITING:
				{
					//another thread is offering to exchange an item, attempt to accept their exchange using our item
					if( m_slot.compareAndSwap(item, STATE_BUSY, slotItem, STATE_EXCHANGE_WAITING) )
					{
						//successful exchange, return the item from the other thread
						*resultItem = slotItem;
						return true;
					}
				}
				break;
			case STATE_BUSY:
				break;
			default:
				assert(false);
			}

			++iterationCount;
		}
		return false;
	}

	template<typename T, unsigned int MAX_ITERATIONS>
	bool LockFreeExchanger<T, MAX_ITERATIONS>::offer(const T& item)
	{
		unsigned int iterationCount = 0;
		while( iterationCount < MAX_ITERATIONS )
		{
			T slotItem;
			State slotState;
			m_slot.get(&slotItem, &slotState);
			switch( slotState )
			{
			case STATE_EMPTY:
				{
					//slot is empty, so attempt to offer our item
					if( m_slot.compareAndSwap(item, STATE_OFFER_WAITING, slotItem, STATE_EMPTY) )
					{
						//wait for another thread to rendezvous and accept our item
						unsigned int waitIterationCount = 0;
						while( waitIterationCount < MAX_ITERATIONS )
						{
							m_slot.get(&slotItem, &slotState);
							if( slotState == STATE_BUSY )
							{
								//offer was successful, another thread has accepted our item
								//no compareAndSwap needed here, we're the only thread that can change the state now
								m_slot.set(T(), STATE_EMPTY);
								return true;
							}
							++waitIterationCount;
						}
						//time's up, attempt to cancel our exchange
						if( m_slot.compareAndSwap(T(), STATE_EMPTY, item, STATE_OFFER_WAITING) )
						{
							//cancelled successfully, return since we've already spent all the iterations available
							return false;
						}
						else
						{
							//cancel failed, that means the offer must have been accepted at the last moment
							assert(m_slot.getSecond()==STATE_BUSY);
							m_slot.set(T(), STATE_EMPTY);
							return true;
						}
					}
				}
				break;
			case STATE_EXCHANGE_WAITING:
			case STATE_OFFER_WAITING:
			case STATE_BUSY:
				break;
			default:
				assert(false);
			}

			++iterationCount;
		}
		return false;
	}

	template<typename T, unsigned int MAX_ITERATIONS>
	bool LockFreeExchanger<T, MAX_ITERATIONS>::accept(T* resultItem)
	{
		unsigned int iterationCount = 0;
		while( iterationCount < MAX_ITERATIONS )
		{
			T slotItem;
			State slotState;
			m_slot.get(&slotItem, &slotState);
			switch( slotState )
			{
			case STATE_OFFER_WAITING:
				{
					//another thread is offering an item, attempt to accept their offer
					if( m_slot.compareAndSwap(T(), STATE_BUSY, slotItem, STATE_OFFER_WAITING) )
					{
						//successful accept, return the item from the other thread
						*resultItem = slotItem;
						return true;
					}
				}
				break;
			case STATE_EMPTY:
			case STATE_EXCHANGE_WAITING:
			case STATE_BUSY:
				break;
			default:
				assert(false);
			}

			++iterationCount;
		}
		return false;
	}
}

#endif
